package org.jetlinks.pro.messaging.kafka.impl;

import org.hswebframework.web.bean.FastBeanCopier;
import org.jetlinks.pro.datasource.DataSource;
import org.jetlinks.pro.datasource.DataSourceConfig;
import org.jetlinks.pro.datasource.DataSourceProvider;
import org.jetlinks.pro.datasource.DataSourceType;
import org.jetlinks.pro.messaging.kafka.KafkaDataSource;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Mono;

import javax.annotation.Nonnull;

@Component
public class KafkaDataSourceProvider implements DataSourceProvider {
    @Nonnull
    @Override
    public DataSourceType getType() {
        return KafkaDataSource.type;
    }

    @Nonnull
    @Override
    public Mono<DataSource> createDataSource(@Nonnull DataSourceConfig properties) {
        KafkaProperties kafkaProperties = FastBeanCopier.copy(properties.getConfiguration(), KafkaProperties::new);
        return Mono.just(create(properties.getId(), kafkaProperties));
    }

    @Nonnull
    @Override
    public Mono<DataSource> reload(@Nonnull DataSource dataSource, @Nonnull DataSourceConfig properties) {
        KafkaProperties rabbitProperties = FastBeanCopier.copy(properties.getConfiguration(), KafkaProperties::new);
        if (dataSource.isWrapperFor(DefaultKafkaDataSource.class)) {
            dataSource
                .unwrap(DefaultKafkaDataSource.class)
                .setConfig(rabbitProperties);
            return Mono.just(dataSource);
        }
        dataSource.dispose();
        return createDataSource(properties);
    }

    public static KafkaDataSource create(String id, KafkaProperties properties) {
        return new DefaultKafkaDataSource(id, properties);
    }
}
